package com.shujia.spark.core

import java.lang

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo21Accumulator {
  def main(args: Array[String]):Unit ={
    val conf: SparkConf = new SparkConf()
      .setAppName("Accumulator")
      .setMaster("local")

    //创建spark对象
    val sc = new SparkContext(conf)

    //新建一个简单的RDD　　
    val RDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9))

    var j = 0
    RDD.foreach(i => {
      j +=1
    })
    println(j)//0
    /**
      * 因为Executor里面的j是一个变量副本，
      * 变量副本更改并不能引起Driver里面变量的更改
      */

    /**
      * 累加器
      * 1、累加器只能Driver定义
      * 2、累加器只能在Executor累加
      * 3、累加器只能在Driver读取
      *
      *
      */

    //1、在Driver端定义累加器
    val accumulator: LongAccumulator = sc.longAccumulator

    RDD.foreach(i =>{
      //2、在Executor端累加
      accumulator.add(i)
    })
    //3、在Driver读取累加结果
    val count: lang.Long = accumulator.value

    println(count)

    /**
      * 累加器的使用场景
      *
      * 如果不使用累加器需要单独启动一个job计算总人数
      * 如果使用累加器，累加计算和班级人数的计算在一起计算出来
      */
    //读取学生信息
    val stuRdd: RDD[String] = sc.textFile("data/students.txt")
    //不使用累加器

    //学生总人数
    val c: Long = stuRdd.count()


    //使用累加器
    //1、定义累加器
    val stuNum: LongAccumulator = sc.longAccumulator

    val kvRDD: RDD[(String, Int)] = stuRdd.map(stu => {
      //累加
      stuNum.add(1)
      val clazz: String = stu.split(",")(4)
      (clazz, 1)
    })


    val clazzRDDNum: RDD[(String, Int)] = kvRDD.reduceByKey(_+_)


    clazzRDDNum.foreach(println)
    //学生总人数
    val studentNum: lang.Long = stuNum.value
    println(studentNum)
  }
}
